

import java.util.*;
public class SourceManager {
	Queue<Request> _requests = null;
	Request _currentRequest = null;
	Request _nextRequest = null;
	//String _peerName = null;
	//ISocket _socket = null;
	DB _db = null;
	//InetAddress _sourceIP = null;
	//int _sourcePort;
	private RequestSource _source = null;
	private long _lastCommunication;
	long _maxIdleTime;
	
	public SourceManager(DB db, Request req, RequestSource source, long maxIdleTime){
		//_socket = socket;
		//_peerName = peerName;
		_maxIdleTime = maxIdleTime;
		_lastCommunication = System.currentTimeMillis();
		_db = db;
		//_sourceIP = req.getRequest().getAddress();
		//_sourcePort = req.getRequest().getPort();
		_source = source;
		_currentRequest = req;
		_nextRequest = null;
		_requests = new LinkedList<Request>();
		requestEntry();
	}
	public RequestSource getSource() { return _source; }
	private void requestEntry(){ _source.getSource().getSocket().send(_currentRequest.getRequest()); }
	
	public void addRequest(Request req){
		// update dbVersion for incremental request
		req.setDbVersion(_currentRequest.getDestinationDbVer());
		Message msg = req.getRequest();
		//msg.setData(("REQUESTENTRY<" + Utils.DateToString(_currentRequest.getDestinationDbVer()) + ">").getBytes());
		msg.setData(("requestentry<$dbversion=" + Utils.DateToString(_currentRequest.getDestinationDbVer()) + ">").getBytes());
		// if no next request is exist
		if (_nextRequest == null){
			// if the new request's dbVersion is updater than the current request dbVersion   
			if (req.getDestinationDbVer().compareTo(_currentRequest.getDestinationDbVer()) == 1)
				_nextRequest = req;		
		}
		else{ // if next request is already exist
			// if the new request's destination date is newer than the existing next request 
			if (req.getDestinationDbVer().compareTo(_nextRequest.getDestinationDbVer()) == 1)
				// update the next request
				_nextRequest = req;
		}
	}
	public boolean entryWasArrived(String peerName, int sn, int n){
				
		// if this is the first entry
		if (_currentRequest.getEntriesExpected() == -1)
			_currentRequest.setEntriesExpected(n);
		
		// if the entry n value doesn't fit to the n value of the current request
		if (_currentRequest.getEntriesExpected() != n)
			return false;
		
		//if serial number is wrong (==0)
		if (sn == 0)
			return false;
		
		// update last communication time
		_lastCommunication = System.currentTimeMillis();
		
		_currentRequest.setBit(sn, peerName);
		
		// if all expected entries was arrived
		if (_currentRequest.getEntriesArrived() == _currentRequest.getEntriesExpected()){
			// if sub-entries are exists
			if (!_requests.isEmpty())
				_currentRequest = _requests.poll();
			// if new request was received during the handling with the first request
			else if (_nextRequest != null){
				_currentRequest = _nextRequest;
				_nextRequest = null;
			}
			else
				// no more requests from the current source
				return true;
			// initialize the new request
			requestEntry();
		}
		// no all of the data expected was received from the source
		return false;
	}
	public int status(){
		boolean sessionWasChanged = false;
		if (System.currentTimeMillis()-_lastCommunication < _maxIdleTime)
			//still receiving
			return 0;
		// assume no data will arrive in the future //
		
		// check if peer is offline
		Record peerRec = _db.getRecord(_source.getPeerName());
		if(peerRec.get_status() == 0)
			// source is not active any more
			return 2;
		
		// generate new sub-requests //
		
		// check if the connection with the source was changed (IP, port, socket);
		Source currentSession = SessionsManager.getSocket(_source.getPeerName());
		if (!_source.getSource().equals(currentSession)){
				_source.setSource(currentSession);
				sessionWasChanged = true;
			}
		// if no entry was arrived
		if (_currentRequest.getEntriesExpected() == -1)
			requestEntry();
		else{ // if part of the entries was arrived 
			generateSubRequests();
			_currentRequest = _requests.poll();
			requestEntry();
		}
		if (sessionWasChanged)
			return 1;
		return 0;
		
	}
	private void generateSubRequests(){
		int i = 1;
		int limit = _currentRequest.getEntriesExpected() + 1;
		String left = null;
		String right = null;
		//int indexLeft, 
		int indexRight;
		//indexLeft = 0;
		indexRight = 1;
		String sEmpty = "-1";
		while (i < limit) {
			String data = null;
			if (!_currentRequest.wasArrived(i).equals(sEmpty)) {
				left = _currentRequest.wasArrived(i);
				//indexLeft = i;
				++i;
			} else {
				//right = left; //Check last index
				int j = i;
				right = null;
				
				//while (j < limit && right != null)
				
				while (j < limit && right == null) {
					if (!_currentRequest.wasArrived(j).equals(sEmpty)) {
						right = _currentRequest.wasArrived(j);
						indexRight = j;
					}
					else 
						++j;
				}
			
				if (right == null) { // now - ALL {
					//data = String.format("REQUESTENTRY<%s,%s,ALL>", Utils.DateToString(_currentRequest.getDbVersion()),left);
					data = String.format("requestentry<$dbversion=%s,$range(%s,ALL)>", Utils.DateToString(_currentRequest.getDbVersion()),left);
					i = limit; // i = j;
					// create new request
					_requests.add(new Request(_currentRequest.getDbVersion(),
							_currentRequest.getDestinationDbVer(),
							new Message(data.getBytes(), _source.getSource().getIP(),_source.getSource().getPort())));
				} else { //left - right
					if (left == null) {
						//SEND with 0 param
						//data = String.format("REQUESTENTRY<%s,0,%s>", Utils.DateToString(_currentRequest.getDbVersion()),right);
						data = String.format("requestentry<$dbversion=%s,$range(0,%s)>", Utils.DateToString(_currentRequest.getDbVersion()),right);
					} else {
						//send left - right
						//data = String.format("REQUESTENTRY<%s,%s-%s>", Utils.DateToString(_currentRequest.getDbVersion()), left, right);
						data = String.format("requestentry<$dbversion=%s,$range(%s-%s)>", Utils.DateToString(_currentRequest.getDbVersion()), left, right);
					}
					// create new request
					_requests.add(new Request(_currentRequest.getDbVersion(),
							_currentRequest.getDestinationDbVer(),
							new Message(data.getBytes(),_source.getSource().getIP(),_source.getSource().getPort())));
					i = indexRight;
				}
			}
			
		}
	}
	
}

